package com.archgeek.zookeeper

import org.apache.curator.framework.CuratorFrameworkFactory
import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode
import org.apache.curator.retry.ExponentialBackoffRetry
import org.apache.curator.utils.ZKPaths
import org.apache.zookeeper.{CreateMode, KeeperException, WatchedEvent, Watcher}
import org.apache.zookeeper.KeeperException.NodeExistsException
import org.apache.zookeeper.data.Stat

import java.nio.charset.StandardCharsets
import java.util.concurrent.TimeUnit

import scala.collection.JavaConverters._
/**
 *
 *
 * Create by pizhihui on 2022-04-29
 */
object ZookeeperKyuubiTest {



  def test(): Unit = {

    val zkStr = "127.0.0.1:2181"

    val client = CuratorFrameworkFactory.builder()
      .connectString(zkStr)
      .connectionTimeoutMs(1000)
      .sessionTimeoutMs(1000)
      .retryPolicy(new ExponentialBackoffRetry(1, 3))
      .build()
    client.start()


    val namespace = "kyuubi"

    val ns = ZKPaths.makePath(null, namespace)
    try {
      client.create()
        .creatingParentsIfNeeded()
        .withMode(CreateMode.PERSISTENT)
        .forPath(ns)
    } catch {
      case _: NodeExistsException =>
      case e: KeeperException =>
        throw new RuntimeException("error: create node", e)
    }


    val server = "127.0.0.1:3451"
    val childPath = ZKPaths.makePath(namespace, s"serverUri=$server;version=3.0.2;sequence=00001")

    val childData = server.getBytes(StandardCharsets.UTF_8)

    try {
      val serviceNode = new PersistentEphemeralNode(client,
        PersistentEphemeralNode.Mode.EPHEMERAL_SEQUENTIAL,
        childPath,
        childData)
      serviceNode.start()
      if(!serviceNode.waitForInitialCreate(10L, TimeUnit.SECONDS)) {
        throw new RuntimeException("create node timeout")
      }

      val znodePath = serviceNode.getActualPath
      val stat: Stat = client.checkExists()
        .usingWatcher(new DeNodeWatcher)
        .forPath(znodePath)
      println(s"znode: ${stat}")


    }catch {
      case _: Exception =>
        println("error: ")
    }


    val getChildPath = client.getChildren.forPath(ns)
    getChildPath.asScala.takeRight(2).map{ p =>
      println(s"ppppppp: ${p}")
      val version = p.split(";").find(_.startsWith("version=")).map(_.stripPrefix("version="))
      val engineRefId = p.split(";").find(_.startsWith("refId=")).map(_.stripPrefix("refId="))
      println(version)
      println(engineRefId)
    }

    System.in.read()

  }


  private class DeNodeWatcher extends Watcher {
    override def process(event: WatchedEvent): Unit = {
      if(event.getType == Watcher.Event.EventType.NodeDeleted) {
        println("节点不存在啦...........")
      }
    }
  }


  def main(args: Array[String]): Unit = {

    test()

  }

}
